package com.autoai.tripflink

import java.util.Properties

import com.autoai.model.{StandCarData, TripData}
import net.sf.json.JSONObject
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.runtime.state.KeyedStateFunction
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011
import org.apache.flink.table.shaded.org.joda.time.Instant
import org.apache.flink.util.Collector


object flinkTrips {


  def main(args: Array[String]): Unit = {

    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "10.30.23.45:9092")
    properties.setProperty("group.id", "guroupuaidi111")
    properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val stream = env.addSource(new FlinkKafkaConsumer011[String]("gwSceneReadDataTest", new SimpleStringSchema(), properties).setStartFromGroupOffsets()).uid("yeah")

    val standData = stream.map(
      (data: String) => {

        val result: JSONObject = JSONObject.fromObject(data)
        val vin = result.get("vin").toString
        val userid = result.get("userId").toString
        val repoerTime = result.get("reportTime").toString.toLong
        val VehTotDistance = result.get("VehTotDistance").toString.toDouble
        val lon = result.get("lon").toString.toDouble
        val lat = result.get("lat").toString.toDouble
        val SystemPowerMod = result.get("SystemPowerMod").toString.toInt

        StandCarData(vin, userid, repoerTime, VehTotDistance, lon, lat, SystemPowerMod)


      }
    ).keyBy(_.vin)
      .process(new MyTripStateFunction(600000L))


//    standData.print()
//    stream.print("hahaha")

    env.execute("gwSceneReadData test")


  }

}

class MyTripStateFunction(l: Long) extends KeyedProcessFunction[String, StandCarData, TripData] {

  val date = "yyyy-MM-dd HH:mm:ss"
  lazy val lastDistace: ValueState[Double] = getRuntimeContext.getState(new ValueStateDescriptor[Double]("distance", classOf[Double]))

  lazy val lastTime: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("timestemp", classOf[Long]))

  lazy val lastTripData: ValueState[TripData] = getRuntimeContext.getState(new ValueStateDescriptor[TripData]("tripdata", classOf[TripData]))


  override def processElement(value: StandCarData, ctx: KeyedProcessFunction[String, StandCarData, TripData]#Context, out: Collector[TripData]): Unit = {

    val lastdic = lastDistace.value()
    val lasttm = lastTime.value()
    var lasttrip = lastTripData.value()

//    println(lastdic)
//    println(lasttrip)
    val diff = value.mileage - lastdic
//    println("distance " + diff)


      lasttrip = TripData(
        tripId = value.vin.reverse + "_" + value.userId + "_" + value.timeStamp, //实时程序传过来的数据就是拼接好的tripid
        vin = value.vin,
        userId = value.userId,
        t_end_mil = value.mileage,
        t_distance =  diff.toInt,
        t_time = 0,
        s_time = value.timeStamp,
        e_time = value.timeStamp,
        s_lon = value.lon,
        s_lat = value.lat,
        e_lon = value.lon,
        e_lat = value.lat,
        c_time = Instant.now().toMutableDateTime.toString(date)
      )


    lastDistace.update(value.mileage)
    lastTripData.update(lasttrip)
    if (diff >= 1) {
      lastTripData.update(lasttrip)
//      lasttrip = updateState(value, lasttrip)
//     lasttrip = getTrip(lasttrip)
println("diff = " + diff)
      out.collect(lasttrip )
    }



  }

  private def updateState(data: StandCarData, lastState: TripData): TripData = {

    //行驶里程数    前后两条数据行驶相差大于10公里，默认为1公里
    var diffMil = 0
    if (Math.abs((data.mileage - lastState.t_end_mil).toInt) > 10) {
      diffMil = 1
    } else {
      diffMil = Math.abs(data.mileage.toInt - lastState.t_end_mil.toInt)
    }

    TripData(
      tripId = lastState.tripId,
      vin = lastState.vin,
      userId = lastState.userId,
      t_end_mil = data.mileage,
      t_distance = lastState.t_distance + diffMil,
      t_time = data.timeStamp - lastState.s_time,
      s_time = lastState.s_time,
      e_time = data.timeStamp,
      s_lon = lastState.s_lon,
      s_lat = lastState.s_lat,
      e_lon = data.lon,
      e_lat = data.lat,
      c_time = Instant.now().toMutableDateTime.toString(date)
    )
  }
  private def getTrip(tripData: TripData): TripData = {

    TripData(
      tripId = tripData.tripId,
      vin = tripData.vin,
      userId = tripData.userId,
      t_end_mil = tripData.t_end_mil,
      t_distance = tripData.t_distance,
      t_time = (tripData.e_time - tripData.s_time) / 1000,
      s_time = tripData.s_time,
      e_time = tripData.e_time,
      s_lon = tripData.s_lon,
      s_lat = tripData.s_lat,
      e_lon = tripData.e_lon,
      e_lat = tripData.e_lat,
      c_time = Instant.now().toMutableDateTime.toString(date)
    )
  }


}
